//package com.atguigu.edu.realtime.app.dws;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.atguigu.edu.realtime.beans.TrafficPageViewBean;
//import com.atguigu.edu.realtime.utils.DateFormatUtil;
//import com.atguigu.edu.realtime.utils.MyKafkaUtil;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.flink.api.common.functions.RichMapFunction;
//import org.apache.flink.api.common.state.StateTtlConfig;
//import org.apache.flink.api.common.state.ValueState;
//import org.apache.flink.api.common.state.ValueStateDescriptor;
//import org.apache.flink.api.common.time.Time;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.datastream.KeyedStream;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
//
///**
// * @author 杨晨昱
// */
//public class DwsTrafficVcChArIsNewPageViewWindow {
//    public static void main(String[] args) {
//
//
//    //TODO 1.基本环境准备
//    //1.1 指定流处理环境
//    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//    //1.2 设置并行度
//       env.setParallelism(4);
//    //TODO 2.检查点相关的设置(略)
//    //TODO 3.从kafka的主题中读取数据
//    //3.1 声明消费的主题以及消费者组
//    String topic = "dwd_traffic_page_log";
//    String groupId = "dws_traffic_vc_ch_ar_isnew_group";
//    //3.2 创建消费者对象
//    FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
//    //3.3 消费数据  封装为流
//    DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
//    //TODO 4.对读取的流中数据进行类型转换 jsonStr->jsonObj
//    SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
//    // jsonObjDS.print(">>>>");
//
//    //TODO 5.按照mid进行分组
//    KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
//
//    //TODO 6.对流中读取的数据进行计数  jsonObj->统计实体类  (使用Flink的状态编程 对uv进行判断)
//    SingleOutputStreamOperator<TrafficPageViewBean> pageViewBeanDS = keyedDS.map(
//            new RichMapFunction<JSONObject, TrafficPageViewBean>() {
//                private ValueState<String> lastVisitDateState;
//
//                @Override
//                public void open(Configuration parameters) throws Exception {
//                    ValueStateDescriptor<String> valueStateDescriptor
//                            = new ValueStateDescriptor<String>("lastVisitDateState", String.class);
//                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
//                    lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
//                }
//
//                @Override
//                public TrafficPageViewBean map(JSONObject jsonObj) throws Exception {
//                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
//                    String vc = commonJsonObj.getString("vc");
//                    String ch = commonJsonObj.getString("ch");
//                    String ar = commonJsonObj.getString("ar");
//                    String isNew = commonJsonObj.getString("is_new");
//
//                    JSONObject pageJsonObj = jsonObj.getJSONObject("page");
//                    String lastPageId = pageJsonObj.getString("last_page_id");
//
//                    //判断是否为独立访客
//                    Long uvCt = 0L;
//                    //获取当前设备上次访问日期
//                    String lastVisitDate = lastVisitDateState.value();
//                    //获取当前访问日期
//                    Long ts = jsonObj.getLong("ts");
//                    String curVisitDate = DateFormatUtil.toDate(ts);
//                    if (StringUtils.isEmpty(lastVisitDate) || !lastVisitDate.equals(curVisitDate)) {
//                        uvCt = 1L;
//                        lastVisitDateState.update(curVisitDate);
//                    }
//                }
//                }
